package com.surfilter.massdata.spark.task;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;

import scala.Tuple2;

import com.act.sparkanalyz.service.impl.SparkService.OutQueueEntity;
import com.act.sparkanalyz.task.ISparkTask;
import com.surfilter.massdata.spark.bean.DnsCount;
import com.surfilter.massdata.spark.bean.SA7602;
import com.surfilter.massdata.spark.util.CommonUtils;
import com.surfilter.massdata.spark.util.DateUtil;
import com.surfilter.massdata.spark.util.URLUtil;

public class DomainAnalyzTask_SA7602 implements ISparkTask{
	
	private static final Log log = LogFactory.getLog(DomainAnalyzTask_SA7602.class);
	private static final long serialVersionUID = 1L;

	@SuppressWarnings("serial")
	@Override
	public List<OutQueueEntity> execute(Map<String, DataFrame> dataFrames, Map<String, String> commandMap) {
		log.info("==============DomainAnalyzTask_SA7602 begin====================");
		List<OutQueueEntity> outList = new ArrayList<OutQueueEntity>();
		try{
			DataFrame dataSource = dataFrames.get("d_dns_domain_hour");
			
			JavaRDD<String> domainSuffix = dataSource.toJavaRDD().map(new Function<Row,String>(){
				private static final long serialVersionUID = 1L;
				@Override
				public String call(Row row){
					String domain = row.getAs("domain");
					String topDomain = "";
					try{
						 topDomain = URLUtil.getDomainName("http://" + domain);
					}catch(Exception e){
						if(log.isErrorEnabled()){
							log.error(e.getMessage(),e);
						}
					}
					return topDomain + "@" + row.getAs("dns_visit_count");
				}});
	        JavaPairRDD<String,Long> domainPair = domainSuffix.mapToPair(new PairFunction<String,String,Long>(){
				private static final long serialVersionUID = 1L;
				@Override
				public Tuple2<String, Long> call(String value){
					try{
						String[] valueStr = value.split("@");
						String domain = valueStr[0];
						if(StringUtils.isNotBlank(domain)){
							long count = Long.parseLong(valueStr[1]);
							
							return new Tuple2<String,Long>(domain,count);
						}
					}catch(Exception e){
						if(log.isErrorEnabled()){
							log.error(e.getMessage(),e);
						}
					}
					
					return new Tuple2<String,Long>("",0L);
				}});
	        JavaPairRDD<String,Long> domainCounts = domainPair.reduceByKey(new Function2<Long,Long,Long>(){
				private static final long serialVersionUID = 1L;
				@Override
				public Long call(Long x, Long y)throws Exception {
					return x+y;
				}});
	        
	        
	        JavaRDD<DnsCount> dnsRDD = domainCounts.map(new Function<Tuple2<String,Long>,DnsCount>() {

				@Override
				public DnsCount call(Tuple2<String, Long> tuple){
					DnsCount sa = new DnsCount();
					try{
						String domain = tuple._1;
						if(StringUtils.isNotBlank(domain)){
							 sa.setTop_domain(domain);
				    		 sa.setCount(tuple._2);
						}
					}catch(Exception e){
						if(log.isErrorEnabled()){
							log.error(e.getMessage(),e);
						}
					}
					return sa;
				}
			});
	    	
	    	DataFrame df_dc = dataSource.sqlContext().createDataFrame(dnsRDD, DnsCount.class);
	        df_dc.registerTempTable("df_dc");
			String sql = "select a.top_domain,a.count, "
					+ "(CASE WHEN ISNULL(b.license_key) = false THEN b.license_key ELSE 'N/A' END) AS license_key,"
					+ "(CASE WHEN ISNULL(b.site_name) = false THEN b.site_name ELSE 'N/A' END) AS site_name "
					+ "from df_dc a left join br2005_temp b on a.top_domain = b.top_domain order by a.count desc limit 100"; 

			DataFrame oper_tmp = dataSource.sqlContext().sql(sql);
			List<Row> top_domain_100 = oper_tmp.toJavaRDD().collect();//把数库中的数据全部输出
		
	    	List<SA7602> list_SA7602 = new ArrayList<SA7602>();	
	        String dayStr = commandMap.get("-d");
	        Date date = DateUtil.getExecDate(DateUtil.getYesterday(),dayStr);
	        if(top_domain_100.size() > 0){
	        	int rank = 1;
	        	for(Row row_top_domain_100 : top_domain_100){
	        	    SA7602 sa = new SA7602();
	        		sa.setYear(DateUtil.getCurrentYear(date));
	        		sa.setHalf_year(DateUtil.getHalfYear(date));
	        		sa.setQuarter(DateUtil.getQuarter(date));
	        		sa.setMonth(DateUtil.getCurrentMonth(date));
	        		sa.setWeek(DateUtil.getCurrentWeek(date));
	        		sa.setDay(DateUtil.getCurrentDay(date));
	        		sa.setBuss_type("WEBSITE_RANK_SUM");
	        		sa.setBuss_value("");
	        		sa.setRank(rank++);
	        		sa.setDomain(row_top_domain_100.getAs("top_domain").toString());
	        	    sa.setWebsite_name(row_top_domain_100.getAs("site_name").toString());
	        	    sa.setIcp_num(row_top_domain_100.getAs("license_key").toString());
	        	    sa.setTraffic(row_top_domain_100.getAs("count").toString());
	        		sa.setSta_range(1);
	        		list_SA7602.add(sa);
	        	}
	        }	
	        df_dc.sqlContext().dropTempTable("df_dc");
	    	JavaSparkContext ctx = new JavaSparkContext(dataSource.sqlContext().sparkContext());
	    	JavaRDD<SA7602> rdd = ctx.parallelize(list_SA7602);
	    	DataFrame df = dataSource.sqlContext().createDataFrame(rdd, SA7602.class);
			outList.add(new OutQueueEntity("SA7602_TEMP", df));
			
			CommonUtils.deleteTaskTableData("SA7602", date, "WEBSITE_RANK_SUM", 1, "day");
		}catch(Exception e){
			if(log.isErrorEnabled()){
				log.error(e.getMessage(),e);
			}
		}
		return outList;
	}
}
